package com.learn.flink.learn.window;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

public class WindowDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
        String hostname="192.168.1.3";
        int port = 7777;
        DataStreamSource<String> data = env.socketTextStream(hostname, port);

        SingleOutputStreamOperator<Tuple1<String>> map = data.map(new MapFunction<String, Tuple1<String>>() {
            @Override
            public Tuple1<String> map(String s) throws Exception {
                return new Tuple1<>(s);
            }
        });
        KeyedStream<Tuple1<String>, Tuple> keyedStream = map.keyBy(0);
        //滚动窗口
//        WindowedStream<Tuple1<String>, Tuple, TimeWindow> win = keyedStream.timeWindow(Time.seconds(10));
        //滑动窗口
//        WindowedStream<Tuple1<String>, Tuple, TimeWindow> win = keyedStream.timeWindow(Time.seconds(10), Time.seconds(5));
        WindowedStream<Tuple1<String>, Tuple, GlobalWindow> win = keyedStream.countWindow(5L);
        SingleOutputStreamOperator<String> result = win.apply(new WindowFunction<Tuple1<String>, String, Tuple, GlobalWindow>() {
            @Override
            public void apply(Tuple tuple, GlobalWindow window, Iterable<Tuple1<String>> input, Collector<String> out) throws Exception {
                for (Tuple1<String> t : input) {
                    out.collect(t.<String>getField(0).toString());
                }
            }
        });
//        //time
//        SingleOutputStreamOperator<String> result = win.apply(new WindowFunction<Tuple1<String>, String, Tuple, TimeWindow>() {
//            @Override
//            public void apply(Tuple tuple, TimeWindow window,
//                              Iterable<Tuple1<String>> input, Collector<String> out) throws Exception {
//                for (Tuple1<String> t : input) {
//                    out.collect(t.getField(0).toString());
//                }
//            }
//        });
        result.print();
        env.execute();
    }
}
